package com.abyss.examples;

import org.apache.commons.collections4.IterableUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.util.Collector;
import java.util.Iterator;
import java.util.List;

/**
 * 监控机架温度
 * 5秒内连续两次上报的温度超过了温度阈值，且后一次的温度超过了前一次的温度，则触发告警
 */
public class SystemMonitoring {
    //定义温度的阈值
    private static final double TEMPPERATURE_THEWHOLD = 100;

    public static void main(String[] args) throws Exception {
        //1：初始化流处理的运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //2：设置按照事件时间处理数据
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //3：并行度不指定，默认线程数跟当前环境的线程数有关（8）
        env.setParallelism(1);

        //开启checkpoint
        env.enableCheckpointing(60000);
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(100000);
        env.getCheckpointConfig().setFailOnCheckpointingErrors(false);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(100);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);

        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.seconds(300), Time.seconds(10)));

        //接入数据源
        DataStreamSource<String> socketTextStream = env.socketTextStream("node01", 9999);

        //将字符串转换成温度事件对象
        SingleOutputStreamOperator<TemperatureEvent> watermarksDataStream = socketTextStream.map(new MapFunction<String, TemperatureEvent>() {
            @Override
            public TemperatureEvent map(String value) throws Exception {
                String[] fields = value.split(",");
                return new TemperatureEvent(Integer.parseInt(fields[0]), Double.parseDouble(fields[1]), Long.parseLong(fields[2]));
            }
        }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<TemperatureEvent>(org.apache.flink.streaming.api.windowing.time.Time.seconds(0)) {
            @Override
            public long extractTimestamp(TemperatureEvent element) {
                return  element.getTimestamp();
            }
        });

        watermarksDataStream.print("上报数据>>");
        watermarksDataStream.keyBy("rackID").process(new TemperatureWarningProcess()).printToErr("警告数据>>");

        //执行任务
        env.execute();
    }

    //警告事件处理
    public static class TemperatureWarningProcess extends KeyedProcessFunction<Tuple, TemperatureEvent, Tuple2<Integer, String>>{
        //定义温度数据集合并保存到状态中
        ListState<TemperatureEvent> temperatureList = null;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            temperatureList = getRuntimeContext().getListState(new ListStateDescriptor<TemperatureEvent>("tempList", TypeInformation.of(new TypeHint<TemperatureEvent>() {})));
        }

        @Override
        public void processElement(TemperatureEvent value, Context ctx, Collector<Tuple2<Integer, String>> out) throws Exception {
            //取出上一次的温度值
            TemperatureEvent lastTemperatureEvent = null;
            Iterable<TemperatureEvent> temperatureWarnings = temperatureList.get();
            Iterator<TemperatureEvent> iterator = temperatureWarnings.iterator();
            while (iterator.hasNext()) {
                lastTemperatureEvent = iterator.next();
            }

            //如果是第一条数据
            if(lastTemperatureEvent == null){
                lastTemperatureEvent = new TemperatureEvent(0, 0, 0);
            }

            //如果温度超过了阈值，则将事件数据放入到状态列表中
            if (value.getTemperature() >= TEMPPERATURE_THEWHOLD && value.getTemperature()>lastTemperatureEvent.getTemperature()) {
                //将当前温度数据添加到温度状态集合中
                TemperatureEvent temperatureWarning = new TemperatureEvent(value.getRackID(), value.getTemperature(), value.getTimestamp());
                temperatureList.add(temperatureWarning);

                //如果定时器的时间戳已经被初始化，则重新绑定定时器
                if (lastTemperatureEvent.getTimestamp() == 0) {
                    Long timerTs = ctx.timerService().currentProcessingTime() + 5000L;
                    ctx.timerService().registerProcessingTimeTimer(timerTs);
                }
            } else {
                //如果没有温度警告的温度数据或者当前数据的温度低于温度阈值，则取消定时器
                if (value.getTemperature() > lastTemperatureEvent.getTemperature()) {
                    //如果温度下降，就取消定时器
                    ctx.timerService().deleteProcessingTimeTimer(lastTemperatureEvent.getTimestamp());
                    //清空警告的数据
                    temperatureList.clear();
                }
            }
        }

        //5秒后定时器触发，输出报警
        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<Integer, String>> out) throws Exception {
            Iterable<TemperatureEvent> temperatureEvents = temperatureList.get();
            List<TemperatureEvent> temperatureEventsList = IterableUtils.toList(temperatureEvents);
            if (temperatureEventsList.size() > 1) {
                out.collect(Tuple2.of(temperatureEventsList.get(0).getRackID(), "五秒内上报温度连续两次超过阈值，且温度在持续升高中"));
            }
            temperatureList.clear();
        }
    }

    public static class TemperatureEvent {
        //机架id
        private int rackID;

        //温度
        private double temperature;

        public int getRackID() {
            return rackID;
        }

        public TemperatureEvent() {
        }

        public void setRackID(int rackID) {
            this.rackID = rackID;
        }

        //时间
        private Long timestamp;

        public TemperatureEvent(int rackID, double temperature, long timestamp) {
            this.rackID = rackID;
            this.temperature = temperature;
            this.timestamp = timestamp;
        }

        public Long getTimestamp() {
            return timestamp;
        }

        public void setTimestamp(Long timestamp) {
            this.timestamp = timestamp;
        }

        public double getTemperature() {
            return temperature;
        }

        public void setTemperature(double temperature) {
            this.temperature = temperature;
        }

        @Override
        public String toString() {
            return "TemperatureEvent{" +
                    "rackID=" + rackID +
                    ", temperature=" + temperature +
                    ", timestamp=" + timestamp +
                    '}';
        }
    }
}